package com.hfzy.ihk.web.freeswitchCall.context;

import com.hfzy.ihk.web.freeswitchCall.constant.PublicConfig;
import com.hfzy.ihk.web.freeswitchCall.support.disruptor.base.DisruptorEvent;
import com.hfzy.ihk.web.freeswitchCall.support.disruptor.exception.DisruptorExceptionHandler;
import com.hfzy.ihk.web.freeswitchCall.support.freeswitch.event.FreeswitchCallOutEvent;
import com.hfzy.ihk.web.freeswitchCall.support.freeswitch.event.FreeswitchCallStateEvent;
import com.hfzy.ihk.web.freeswitchCall.support.freeswitch.event.FreeswitchRecordFileEvent;
import com.hfzy.ihk.web.freeswitchCall.support.freeswitch.factory.FreeswitchCallOutEventFactory;
import com.hfzy.ihk.web.freeswitchCall.support.freeswitch.factory.FreeswitchCallStateFactory;
import com.hfzy.ihk.web.freeswitchCall.support.freeswitch.factory.FreeswitchRecordFileEventFactory;
import com.hfzy.ihk.web.freeswitchCall.support.freeswitch.handler.FreeswitchCallOutHandler;
import com.hfzy.ihk.web.freeswitchCall.support.freeswitch.handler.FreeswitchCallStateHandler;
import com.hfzy.ihk.web.freeswitchCall.support.freeswitch.handler.FreeswitchRecordFileHandler;
import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.core.task.TaskExecutor;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * Created by Administrator on 16-5-28.
 */

@Component("disruptorInitBean")
public class DisruptorInitBean implements InitializingBean, DisposableBean {


    //freeswitch状态队列
    public static RingBuffer<FreeswitchCallStateEvent>[] ringBufferFreeswitchCallState=new RingBuffer[PublicConfig.ALIYUN_CALL_STATE_BUFFER_NUM_SIZE];

    //Freeswitch呼出队列
    private static RingBuffer<FreeswitchCallOutEvent> ringBufferFreeswitchCallOut;

    //freeswitch录音接受文件队列
    private static RingBuffer<FreeswitchRecordFileEvent> ringBufferFreeswitchRecordFile;

    private Logger logger = LoggerFactory.getLogger(DisruptorInitBean.class);


    private Disruptor<FreeswitchCallOutEvent> disruptorFreeswitchCallOut;

    private Disruptor<FreeswitchCallStateEvent> disruptorFreeswitchStateCallOut;

    private Disruptor<FreeswitchRecordFileEvent> disruptorFreeswitchRecordFile;




    //线程
    private ExecutorService executorFreeswitchCallOut;
    private ExecutorService executorFreeswitchCallState;
    private ExecutorService executorFreeswitchRecordFile;



    public static RingBuffer<FreeswitchCallStateEvent> getRingBufferFreeswitchCallState(int num){
        return  ringBufferFreeswitchCallState[num];
    }

    public static RingBuffer<FreeswitchCallOutEvent> getRingBufferFreeswitchCallOut() {
        return ringBufferFreeswitchCallOut;
    }

    public static RingBuffer<FreeswitchRecordFileEvent> getRingBufferFreeswitchRecordFile() {
        return ringBufferFreeswitchRecordFile;
    }
    
    @Override
    public void afterPropertiesSet() throws Exception {
        logger.info("开始初始化云呼队列");
        //freeswitch
        initDisruptorFreeswitchCallOut();
        initDisruptorFreeswitchCallState();
        initDisruptorFreeswitchRecordFile();
        //双呼
        initDisruptorBatchInsert();//必须先初始化

        logger.info("初始化云呼队列完成");
    }

    /**
     * 初始化录音队列
     */
    private void initDisruptorFreeswitchRecordFile() {

        int handlerSize = PublicConfig.ALIYUN_RECORD_FILE_HANDLER_SIZE;
        int bufferSize = PublicConfig.ALIYUN_RECORD_FILE_BUFFER_SIZE;

        executorFreeswitchRecordFile = Executors.newFixedThreadPool(handlerSize);

        FreeswitchRecordFileEventFactory callEventFactory = new FreeswitchRecordFileEventFactory();

        disruptorFreeswitchRecordFile = new Disruptor(
                callEventFactory,
                bufferSize,
                executorFreeswitchRecordFile,
                ProducerType.MULTI,
                new BlockingWaitStrategy()
        );

        FreeswitchRecordFileHandler handlers[] = new FreeswitchRecordFileHandler[handlerSize];
        for(int i=0;i<handlerSize;i++){
            handlers[i] = BeanUtils.getFreeswitchRecordFileHandler();
        }

        disruptorFreeswitchRecordFile.handleEventsWithWorkerPool(handlers);

        disruptorFreeswitchRecordFile.handleExceptionsWith(new DisruptorExceptionHandler());

        ringBufferFreeswitchRecordFile = disruptorFreeswitchRecordFile.start();

        logger.info("init disruptorFreeswitchRecordFile over: handler size:{},buffer size:{}",handlerSize,bufferSize);

    }




    /**
     * 初始化freeswitch外呼队列
     */
    private void initDisruptorFreeswitchCallOut(){
        int handlerSize = PublicConfig.ALIYUN_CALL_OUT_HANDLER_SIZE;
        int bufferSize = PublicConfig.ALIYUN_CALL_OUT_BUFFER_SIZE;

        executorFreeswitchCallOut = Executors.newFixedThreadPool(handlerSize);

        FreeswitchCallOutEventFactory callEventFactory = new FreeswitchCallOutEventFactory();

        disruptorFreeswitchCallOut = new Disruptor(
                callEventFactory,
                bufferSize,
                executorFreeswitchCallOut,
                ProducerType.MULTI,
                new BlockingWaitStrategy()
        );

        FreeswitchCallOutHandler[] handlers = new FreeswitchCallOutHandler[handlerSize];
        for (int i = 0; i < handlerSize; i++) {
            handlers[i] = BeanUtils.getFreeswitchCallOutHandler();
        }

        disruptorFreeswitchCallOut.handleEventsWithWorkerPool(handlers);

        disruptorFreeswitchCallOut.setDefaultExceptionHandler(new DisruptorExceptionHandler());

        ringBufferFreeswitchCallOut = disruptorFreeswitchCallOut.start();

        logger.info("init disruptorFreeswitchCallOut over: handler size:{},buffer size:{}", handlerSize, bufferSize);
    }


    /**
     * 初始化freeswitch状态队列
     */
    private void initDisruptorFreeswitchCallState(){

        int handlerSize=PublicConfig.ALIYUN_CALL_STATE_HANDLER_SIZE;

        int bufferSize=PublicConfig.ALIYUN_CALL_STATE_BUFFER_SIZE;

        int bufferNumSize = PublicConfig.ALIYUN_CALL_STATE_BUFFER_NUM_SIZE;

        for(int a=0;a<bufferNumSize;a++){

            executorFreeswitchCallState=Executors.newFixedThreadPool(handlerSize);

            FreeswitchCallStateFactory factory=new FreeswitchCallStateFactory();

            disruptorFreeswitchStateCallOut=new Disruptor(factory,
                    bufferSize,
                    executorFreeswitchCallState,
                    ProducerType.MULTI,new BlockingWaitStrategy());

            FreeswitchCallStateHandler[] handlers=new FreeswitchCallStateHandler[handlerSize];

            for (int b=0;b<handlers.length;b++){

                handlers[b]=BeanUtils.getFreeswitchCallStateHandler();
            }

            disruptorFreeswitchStateCallOut.handleEventsWithWorkerPool(handlers);

            disruptorFreeswitchStateCallOut.setDefaultExceptionHandler(new DisruptorExceptionHandler());

            ringBufferFreeswitchCallState[a]=disruptorFreeswitchStateCallOut.start();
        }

        logger.info("init disruptorFreeswitchCallState over: handler size:{},buffer size:{}", handlerSize, bufferSize);


    }

    /*单消费者*/
    private void initDisruptorBatchInsert() {

    }

    @Override
    public void destroy() throws Exception {
        logger.info("disruptor destroy..............");
        //================================
//        disruptorBatchInsert.shutdown();
//        executorBatchInsert.shutdown();
//
//
//        executorAliyunCallOut.shutdown();
//        executorAliyunCallState.shutdown();
//        executorAliyunRecordFile.shutdown();

        //关闭freeswitch相关
        disruptorFreeswitchStateCallOut.shutdown();
        disruptorFreeswitchCallOut.shutdown();
        disruptorFreeswitchRecordFile.shutdown();
        executorFreeswitchCallState.shutdown();
        executorFreeswitchCallOut.shutdown();
        executorFreeswitchRecordFile.shutdown();
    }

}
